package cn.dmrliu.edu.realtime.app.dws;

import cn.dmrliu.edu.realtime.app.func.DimAsyncFunction;
import cn.dmrliu.edu.realtime.bean.CourseReviewBean;
import cn.dmrliu.edu.realtime.util.ClickhouseUtil;
import cn.dmrliu.edu.realtime.util.DateUtil;
import cn.dmrliu.edu.realtime.util.KafkaUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class DwsInteractionReviewWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1. 基本环境的准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点的设置

        // TODO 3. 从kafka的 评价事务表 获取数据
        String topic = "edu_dwd_interaction_course_review";
        String group = "edu_ads_interaction_course_window_group";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, group);
        DataStreamSource<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");

        // TODO 4. 转为json结构
        SingleOutputStreamOperator<JSONObject> jsonObjectDS = kafkaDS.map(JSON::parseObject);

        // TODO 5. 设置水位线
        SingleOutputStreamOperator<JSONObject> watermarkDS = jsonObjectDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return DateUtil.toTs(element.getString("create_time"), true);
                            }
                        })
        );

        // TODO 6. 按课程分组，开窗，聚合
        SingleOutputStreamOperator<CourseReviewBean> aggregateDS = watermarkDS
                .keyBy(new KeySelector<JSONObject, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> getKey(JSONObject value) throws Exception {
                        return Tuple2.of(value.getString("course_id"), value.getInteger("review_stars"));
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .aggregate(new AggregateFunction<JSONObject, Long, Long>() {
                               @Override
                               public Long createAccumulator() {
                                   return 0L;
                               }

                               @Override
                               public Long add(JSONObject value, Long accumulator) {
                                   return accumulator + 1L;
                               }

                               @Override
                               public Long getResult(Long accumulator) {
                                   return accumulator;
                               }

                               @Override
                               public Long merge(Long a, Long b) {
                                   return null;
                               }
                           },
                        new WindowFunction<Long, CourseReviewBean, Tuple2<String, Integer>, TimeWindow>() {
                            @Override
                            public void apply(Tuple2<String, Integer> key, TimeWindow window, Iterable<Long> input, Collector<CourseReviewBean> out) throws Exception {
                                String sst = DateUtil.toYmdHms(window.getStart());
                                String edt = DateUtil.toYmdHms(window.getEnd());
                                out.collect(new CourseReviewBean(
                                        sst,
                                        edt,
                                        key.f0,
                                        "",
                                        key.f1,
                                        input.iterator().next(),
                                        System.currentTimeMillis()
                                ));
                            }
                        });

//        aggregateDS.print(">>>>>>>");


        // TODO 8. 补充维度信息
        SingleOutputStreamOperator<CourseReviewBean> withCourseNameDS = AsyncDataStream.unorderedWait(
                aggregateDS,
                new DimAsyncFunction<CourseReviewBean>("dim_course_info") {
                    @Override
                    public String getKey(CourseReviewBean courseReviewBean) {
                        return courseReviewBean.getCourseId();
                    }

                    @Override
                    public void join(CourseReviewBean courseReviewBean, JSONObject dimInfo) {
                        String courseName = dimInfo.getString("course_name".toUpperCase());
                        courseReviewBean.setCourseName(courseName);
                    }
                },
                10,
                TimeUnit.SECONDS
        );

        withCourseNameDS.print(">>>>>>");

        // TODO 9. 写入clickhouse中
        withCourseNameDS.addSink(ClickhouseUtil.getSinkFunction("insert into dws_interaction_review_window values(?,?,?,?,?,?,?)"));

        env.execute();
    }
}
